# -*- coding: utf-8 -*-

# Copyright (c) 2017 Ansible, Inc.
# All Rights Reserved.

from contextlib import contextmanager
import codecs
from uuid import uuid4
import time

from unittest import mock

from django.conf import LazySettings
from django.core.cache.backends.locmem import LocMemCache
from django.core.exceptions import ImproperlyConfigured
from django.db.utils import Error as DBError
from django.utils.translation import gettext_lazy as _
import pytest

from awx.conf import models, fields
from awx.conf.settings import SettingsWrapper, EncryptedCacheProxy, SETTING_CACHE_NOTSET
from awx.conf.registry import SettingsRegistry

from awx.main.utils import encrypt_field, decrypt_field


@contextmanager
def apply_patches(_patches):
    [p.start() for p in _patches]
    yield
    [p.stop() for p in _patches]


@pytest.fixture()
def settings(request):
    """
    This fixture initializes a Django settings object that wraps our
    `awx.conf.settings.SettingsWrapper` and passes it as an argument into the
    test function.

    This mimics the work done by `awx.conf.settings.SettingsWrapper.initialize`
    on `django.conf.settings`.
    """
    cache = LocMemCache(str(uuid4()), {})  # make a new random cache each time
    settings = LazySettings()
    registry = SettingsRegistry(settings)
    defaults = {}

    # @pytest.mark.defined_in_file can be used to mark specific setting values
    # as "defined in a settings file".  This is analogous to manually
    # specifying a setting on the filesystem (e.g., in a local_settings.py in
    # development, or in /etc/tower/conf.d/<something>.py)
    for marker in request.node.own_markers:
        if marker.name == 'defined_in_file':
            defaults = marker.kwargs

    defaults['DEFAULTS_SNAPSHOT'] = {}
    settings.configure(**defaults)
    settings._wrapped = SettingsWrapper(settings._wrapped, cache, registry)
    return settings


@pytest.mark.defined_in_file(DEBUG=True)
def test_unregistered_setting(settings):
    "native Django settings are not stored in DB, and aren't cached"
    assert settings.DEBUG is True
    assert settings.cache.get('DEBUG') is None


def test_read_only_setting(settings):
    settings.registry.register('AWX_READ_ONLY', field_class=fields.CharField, category=_('System'), category_slug='system', default='NO-EDITS', read_only=True)
    assert settings.AWX_READ_ONLY == 'NO-EDITS'
    assert len(settings.registry.get_registered_settings(read_only=False)) == 0
    settings = settings.registry.get_registered_settings(read_only=True)
    assert settings == ['AWX_READ_ONLY']


@pytest.mark.defined_in_file(AWX_SOME_SETTING='DEFAULT')
@pytest.mark.parametrize('read_only', [True, False])
def test_setting_defined_in_file(settings, read_only):
    kwargs = {'read_only': True} if read_only else {}
    settings.registry.register('AWX_SOME_SETTING', field_class=fields.CharField, category=_('System'), category_slug='system', **kwargs)
    assert settings.AWX_SOME_SETTING == 'DEFAULT'
    assert len(settings.registry.get_registered_settings(read_only=False)) == 0
    settings = settings.registry.get_registered_settings(read_only=True)
    assert settings == ['AWX_SOME_SETTING']


@pytest.mark.defined_in_file(AWX_SOME_SETTING='DEFAULT')
def test_setting_defined_in_file_with_empty_default(settings):
    settings.registry.register('AWX_SOME_SETTING', field_class=fields.CharField, category=_('System'), category_slug='system', default='')
    assert settings.AWX_SOME_SETTING == 'DEFAULT'
    assert len(settings.registry.get_registered_settings(read_only=False)) == 0
    settings = settings.registry.get_registered_settings(read_only=True)
    assert settings == ['AWX_SOME_SETTING']


@pytest.mark.defined_in_file(AWX_SOME_SETTING='DEFAULT')
def test_setting_defined_in_file_with_specific_default(settings):
    settings.registry.register('AWX_SOME_SETTING', field_class=fields.CharField, category=_('System'), category_slug='system', default=123)
    assert settings.AWX_SOME_SETTING == 'DEFAULT'
    assert len(settings.registry.get_registered_settings(read_only=False)) == 0
    settings = settings.registry.get_registered_settings(read_only=True)
    assert settings == ['AWX_SOME_SETTING']


@pytest.mark.defined_in_file(AWX_SOME_SETTING='DEFAULT')
def test_read_only_defaults_are_cached(settings):
    "read-only settings are stored in the cache"
    settings.registry.register('AWX_SOME_SETTING', field_class=fields.CharField, category=_('System'), category_slug='system')
    assert settings.AWX_SOME_SETTING == 'DEFAULT'
    assert settings.cache.get('AWX_SOME_SETTING') == 'DEFAULT'


@pytest.mark.defined_in_file(AWX_SOME_SETTING='DEFAULT')
def test_cache_respects_timeout(settings):
    "only preload the cache every SETTING_CACHE_TIMEOUT settings"
    settings.registry.register('AWX_SOME_SETTING', field_class=fields.CharField, category=_('System'), category_slug='system')

    assert settings.AWX_SOME_SETTING == 'DEFAULT'
    cache_expiration = settings.cache.get('_awx_conf_preload_expires')
    assert cache_expiration > time.time()

    assert settings.AWX_SOME_SETTING == 'DEFAULT'
    assert settings.cache.get('_awx_conf_preload_expires') == cache_expiration


def test_default_setting(settings, mocker):
    "settings that specify a default are inserted into the cache"
    settings.registry.register('AWX_SOME_SETTING', field_class=fields.CharField, category=_('System'), category_slug='system', default='DEFAULT')

    settings_to_cache = mocker.Mock(**{'order_by.return_value': []})
    mocker.patch('awx.conf.models.Setting.objects.filter', return_value=settings_to_cache)
    assert settings.AWX_SOME_SETTING == 'DEFAULT'
    assert settings.cache.get('AWX_SOME_SETTING') == 'DEFAULT'


@pytest.mark.defined_in_file(AWX_SOME_SETTING='DEFAULT')
def test_setting_is_from_setting_file(settings, mocker):
    settings.registry.register('AWX_SOME_SETTING', field_class=fields.CharField, category=_('System'), category_slug='system')
    assert settings.AWX_SOME_SETTING == 'DEFAULT'
    assert settings.registry.get_setting_field('AWX_SOME_SETTING').defined_in_file is True


def test_setting_is_not_from_setting_file(settings, mocker):
    settings.registry.register('AWX_SOME_SETTING', field_class=fields.CharField, category=_('System'), category_slug='system', default='DEFAULT')

    settings_to_cache = mocker.Mock(**{'order_by.return_value': []})
    mocker.patch('awx.conf.models.Setting.objects.filter', return_value=settings_to_cache)
    assert settings.AWX_SOME_SETTING == 'DEFAULT'
    assert settings.registry.get_setting_field('AWX_SOME_SETTING').defined_in_file is False


def test_empty_setting(settings, mocker):
    "settings with no default and no defined value are not valid"
    settings.registry.register('AWX_SOME_SETTING', field_class=fields.CharField, category=_('System'), category_slug='system')

    mocks = mocker.Mock(**{'order_by.return_value': mocker.Mock(**{'__iter__': lambda self: iter([]), 'first.return_value': None})})
    mocker.patch('awx.conf.models.Setting.objects.filter', return_value=mocks)
    with pytest.raises(AttributeError):
        settings.AWX_SOME_SETTING
    assert settings.cache.get('AWX_SOME_SETTING') == SETTING_CACHE_NOTSET


def test_setting_from_db(settings, mocker):
    "settings can be loaded from the database"
    settings.registry.register('AWX_SOME_SETTING', field_class=fields.CharField, category=_('System'), category_slug='system', default='DEFAULT')

    setting_from_db = mocker.Mock(key='AWX_SOME_SETTING', value='FROM_DB')
    mocks = mocker.Mock(**{'order_by.return_value': mocker.Mock(**{'__iter__': lambda self: iter([setting_from_db]), 'first.return_value': setting_from_db})})
    mocker.patch('awx.conf.models.Setting.objects.filter', return_value=mocks)
    assert settings.AWX_SOME_SETTING == 'FROM_DB'
    assert settings.cache.get('AWX_SOME_SETTING') == 'FROM_DB'


@pytest.mark.defined_in_file(AWX_SOME_SETTING='DEFAULT')
def test_read_only_setting_assignment(settings):
    "read-only settings cannot be overwritten"
    settings.registry.register('AWX_SOME_SETTING', field_class=fields.CharField, category=_('System'), category_slug='system')
    assert settings.AWX_SOME_SETTING == 'DEFAULT'
    with pytest.raises(ImproperlyConfigured):
        settings.AWX_SOME_SETTING = 'CHANGED'
    assert settings.AWX_SOME_SETTING == 'DEFAULT'


def test_db_setting_create(settings, mocker):
    "settings are stored in the database when set for the first time"
    settings.registry.register('AWX_SOME_SETTING', field_class=fields.CharField, category=_('System'), category_slug='system')

    setting_list = mocker.Mock(**{'order_by.return_value.first.return_value': None})
    with apply_patches(
        [
            mocker.patch('awx.conf.models.Setting.objects.filter', return_value=setting_list),
            mocker.patch('awx.conf.models.Setting.objects.create', mocker.Mock()),
        ]
    ):
        settings.AWX_SOME_SETTING = 'NEW-VALUE'

    models.Setting.objects.create.assert_called_with(key='AWX_SOME_SETTING', user=None, value='NEW-VALUE')


def test_db_setting_update(settings, mocker):
    "settings are updated in the database when their value changes"
    settings.registry.register('AWX_SOME_SETTING', field_class=fields.CharField, category=_('System'), category_slug='system')

    existing_setting = mocker.Mock(key='AWX_SOME_SETTING', value='FROM_DB')
    setting_list = mocker.Mock(**{'order_by.return_value.first.return_value': existing_setting})
    mocker.patch('awx.conf.models.Setting.objects.filter', return_value=setting_list)
    settings.AWX_SOME_SETTING = 'NEW-VALUE'

    assert existing_setting.value == 'NEW-VALUE'
    existing_setting.save.assert_called_with(update_fields=['value'])


def test_db_setting_deletion(settings, mocker):
    "settings are auto-deleted from the database"
    settings.registry.register('AWX_SOME_SETTING', field_class=fields.CharField, category=_('System'), category_slug='system')

    existing_setting = mocker.Mock(key='AWX_SOME_SETTING', value='FROM_DB')
    mocker.patch('awx.conf.models.Setting.objects.filter', return_value=[existing_setting])
    del settings.AWX_SOME_SETTING

    assert existing_setting.delete.call_count == 1


@pytest.mark.defined_in_file(AWX_SOME_SETTING='DEFAULT')
def test_read_only_setting_deletion(settings):
    "read-only settings cannot be deleted"
    settings.registry.register('AWX_SOME_SETTING', field_class=fields.CharField, category=_('System'), category_slug='system')
    assert settings.AWX_SOME_SETTING == 'DEFAULT'
    with pytest.raises(ImproperlyConfigured):
        del settings.AWX_SOME_SETTING
    assert settings.AWX_SOME_SETTING == 'DEFAULT'


def test_charfield_properly_sets_none(settings, mocker):
    "see: https://github.com/ansible/ansible-tower/issues/5322"
    settings.registry.register('AWX_SOME_SETTING', field_class=fields.CharField, category=_('System'), category_slug='system', allow_null=True)

    setting_list = mocker.Mock(**{'order_by.return_value.first.return_value': None})
    with apply_patches(
        [
            mocker.patch('awx.conf.models.Setting.objects.filter', return_value=setting_list),
            mocker.patch('awx.conf.models.Setting.objects.create', mocker.Mock()),
        ]
    ):
        settings.AWX_SOME_SETTING = None

    models.Setting.objects.create.assert_called_with(key='AWX_SOME_SETTING', user=None, value=None)


def test_settings_use_cache(settings, mocker):
    settings.registry.register('AWX_VAR', field_class=fields.CharField, category=_('System'), category_slug='system')
    settings.cache.set('AWX_VAR', 'foobar')
    settings.cache.set('_awx_conf_preload_expires', 100)
    # Will fail test if database is used
    getattr(settings, 'AWX_VAR')


def test_settings_use_an_encrypted_cache(settings, mocker):
    settings.registry.register('AWX_ENCRYPTED', field_class=fields.CharField, category=_('System'), category_slug='system', encrypted=True)
    assert isinstance(settings.cache, EncryptedCacheProxy)
    assert settings.cache.__dict__['encrypter'] == encrypt_field
    assert settings.cache.__dict__['decrypter'] == decrypt_field
    settings.cache.set('AWX_ENCRYPTED_ID', 402)
    settings.cache.set('AWX_ENCRYPTED', 'foobar')
    settings.cache.set('_awx_conf_preload_expires', 100)
    # Will fail test if database is used
    getattr(settings, 'AWX_ENCRYPTED')


def test_sensitive_cache_data_is_encrypted(settings, mocker):
    "fields marked as `encrypted` are stored in the cache with encryption"
    settings.registry.register('AWX_ENCRYPTED', field_class=fields.CharField, category=_('System'), category_slug='system', encrypted=True)

    def rot13(obj, attribute):
        assert obj.pk == 123
        return codecs.encode(getattr(obj, attribute), 'rot_13')

    native_cache = LocMemCache(str(uuid4()), {})
    cache = EncryptedCacheProxy(native_cache, settings.registry, encrypter=rot13, decrypter=rot13)
    # Insert the setting value into the database; the encryption process will
    # use its primary key as part of the encryption key
    setting_from_db = mocker.Mock(pk=123, key='AWX_ENCRYPTED', value='SECRET!')
    mocks = mocker.Mock(**{'order_by.return_value': mocker.Mock(**{'__iter__': lambda self: iter([setting_from_db]), 'first.return_value': setting_from_db})})
    mocker.patch('awx.conf.models.Setting.objects.filter', return_value=mocks)
    cache.set('AWX_ENCRYPTED', 'SECRET!')
    assert cache.get('AWX_ENCRYPTED') == 'SECRET!'
    assert native_cache.get('AWX_ENCRYPTED') == 'FRPERG!'


def test_readonly_sensitive_cache_data_is_encrypted(settings):
    "readonly fields marked as `encrypted` are stored in the cache with encryption"
    settings.registry.register('AWX_ENCRYPTED', field_class=fields.CharField, category=_('System'), category_slug='system', read_only=True, encrypted=True)

    def rot13(obj, attribute):
        assert obj.pk is None
        return codecs.encode(getattr(obj, attribute), 'rot_13')

    native_cache = LocMemCache(str(uuid4()), {})
    cache = EncryptedCacheProxy(native_cache, settings.registry, encrypter=rot13, decrypter=rot13)
    cache.set('AWX_ENCRYPTED', 'SECRET!')
    assert cache.get('AWX_ENCRYPTED') == 'SECRET!'
    assert native_cache.get('AWX_ENCRYPTED') == 'FRPERG!'


@pytest.mark.defined_in_file(AWX_VAR='DEFAULT')
def test_in_memory_cache_only_for_registered_settings(settings):
    "Test that we only make use of the in-memory TTL cache for registered settings"
    settings._awx_conf_memoizedcache.clear()
    settings.MIDDLEWARE
    assert len(settings._awx_conf_memoizedcache) == 0  # does not cache MIDDLEWARE
    settings.registry.register('AWX_VAR', field_class=fields.CharField, category=_('System'), category_slug='system')
    settings._wrapped.__dict__['all_supported_settings'] = ['AWX_VAR']  # because it is cached_property
    settings._awx_conf_memoizedcache.clear()
    assert settings.AWX_VAR == 'DEFAULT'
    assert len(settings._awx_conf_memoizedcache) == 1  # caches registered settings


@pytest.mark.defined_in_file(AWX_VAR='DEFAULT')
def test_in_memory_cache_works(settings):
    settings._awx_conf_memoizedcache.clear()
    settings.registry.register('AWX_VAR', field_class=fields.CharField, category=_('System'), category_slug='system')
    settings._wrapped.__dict__['all_supported_settings'] = ['AWX_VAR']

    settings._awx_conf_memoizedcache.clear()

    with mock.patch('awx.conf.settings.SettingsWrapper._get_local', return_value='DEFAULT') as mock_get:
        assert settings.AWX_VAR == 'DEFAULT'
        mock_get.assert_called_once_with('AWX_VAR')

    with mock.patch.object(settings, '_get_local') as mock_get:
        assert settings.AWX_VAR == 'DEFAULT'
        mock_get.assert_not_called()


@pytest.mark.defined_in_file(AWX_VAR=[])
def test_getattr_with_database_error(settings):
    """
    If a setting is defined via the registry and has a null-ish default which is not None
    then referencing that setting during a database outage should give that default
    this is regression testing for a bug where it would return None
    """
    settings.registry.register('AWX_VAR', field_class=fields.StringListField, default=[], category=_('System'), category_slug='system')
    settings._awx_conf_memoizedcache.clear()

    with mock.patch('django.db.backends.base.base.BaseDatabaseWrapper.ensure_connection') as mock_ensure:
        mock_ensure.side_effect = DBError('for test')
        assert settings.AWX_VAR == []
